Can we make this section kafka / avro agnostic? If we just set asBytes() it should work right? Another option is to verify the contentType in which case if it's not json but explicitly sth else then we just pass through? Wdyt?

I also sense that we could have some extension points here. Like check through spi (we already do it in other parts of scc) some interfaces to verify if this contract has a special way of treating payload. We could have some AvroContractPayloadProceesor that would activate when the metadata has avro and it would provide a function to how convert the payload. That way this core logic stays clean of avro but we can inject the behavior. There would have to be some priority / ordering like we do in other spis here in scc.

Wdyt?

Can wIf we just set asBytes() it should work right?

by passing bytes, do you mean the 2nd approach demo-ed here?

If so - yes, it works, but it becomes very hard to peer review the contract changes and understand test failures.

In fact, if we stick to that approach, we won't need any of this - as it mainly covers workarounds needed for the 1st apporach described in the samples PR.

Another option is to verify the contentType in which case if it's not json but explicitly sth else then we just pass through? Wdyt?

Are you suggesting setting a header like contentType: application/avro and relying on it here, instead of the new custom field we added to the contract metadata?

It would force users to add the contentType header to the contract even if they don't use it in prod (as it is not strictly required, more like a convention some are using).

This would work - but imo sounds a bit sketchy. wdyt?
What's your concern about using the contract meta here?

I also sense that we could have some extension points here

Yes, that's a great idea, actually. We have a list of processors, each of them will provide a predicate to test the contract (therefore, we see if we should apply it) , and a function for mapping/extracting the payload.

This way, we can keep using the metadata approach, but extract it somewhere else and keep this agnostic of avro.

+You mentioned potentially adding prtobuff later, which might use this extension point too.

What headers are being sent in case of avro messages? Is there any content type?

afaik, there are no headers strictly required by the kafka broker or the Avro/confluent deserializer — the schema info is embedded in the payload. contentType is sometimes added by convention but isn't enforced by anything.

This X-Correlation-Id header in the test is just a custom header used to verify that user-defined headers are propagated correctly. Would you prefer to rename it to something more obviously test-scoped (e.g. X-Dummy-Custom-Header), or add a contentType: application/avro ?

Add a return

done

Why can't we reuse the users production template?

I was thinking about it: this is meant to be used on the consumer side - therefore we can have a kafkaTemplate configured differently here (eg: seems unlikely, but we can have an app consuming avro and publishing json)

What do you think about looking for a bean named avroKafkaTemplate first, if not, fall back to the "prod" kafkaTempalte bean, if not present either, create one.

 if (ctx.containsBean("avroKafkaTemplate")) {
  return (KafkaTemplate<String, Object>) ctx.getBean("avroKafkaTemplate");
}
if (ctx.containsBean("kafkaTemplate")) {
  return (KafkaTemplate<String, Object>) ctx.getBean("kafkaTemplate");
}
return createAvroKafkaTemplate(bootstrapServers, schemaRegistryUrl);

My main point is to allow overriding this avroKafkaTemplate used by the stub without altering the prod bean.
What do you think about smth like this?

Can you provide a javadoc why we need this?

done